package com.atguigu.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.utils.DruidDSUtil;
import com.atguigu.utils.ThreadPoolUtil;
import lombok.SneakyThrows;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;


import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

//dws09
//public class DimAsyncJoinFunction extends RichAsyncFunction<TradeSkuOrderBean,TradeSkuOrderBean> {
public abstract class DimAsyncJoinFunction<T> extends RichAsyncFunction<T, T> implements JoinFunction<T>{
    private String tableName;
    private DruidDataSource druidDataSource;
    private ThreadPoolExecutor threadPoolExecutor;

    public DimAsyncJoinFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //获得连接池
        druidDataSource = DruidDSUtil.getDruidDataSource();
        //获得线程池
        threadPoolExecutor = ThreadPoolUtil.getThreadPoolExecutor();


    }

    //实现接口implements JoinFunction<T>即可
//    //todo 如果写的是泛型方法，通过传参；如果写的是泛型类：1.构造器传参2.抽象方法
//    //外面调用者必须要调用时必须要重写该方法
//    public abstract String getKey(T input);//不好写的封装到方法里，谁调用这个类，谁就要重写该方法
//    public abstract void join(T input, JSONObject dimInfo);


    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) {//input就是一个一个来的数据，可以时javabean
        threadPoolExecutor.execute(new Runnable() {
            @SneakyThrows
            @Override
            public void run() {
                DruidPooledConnection connection = druidDataSource.getConnection();//通过druid获得phoenix连接，然后将连接传参给jdbc，再通过jdbc.querylist(...)查询phoenix数据

                //todo 查询维表数据
                String key=getKey(input);
                JSONObject dimInfo = DimHandler.getDimInfo(connection, tableName, key);

                //todo 将信息补充到javabean中
                if(dimInfo != null){//有可能没查到
                    join(input,dimInfo);//查询结果不为null的话就将查询结果写到input里
                }

                //todo 写回流里
//                resultFuture.complete(Collections.singletonList(input));//查询结果只有一条数据
                resultFuture.complete(Collections.singleton(input));//查询结果只有一条数据

                //todo 回收phoenix连接
                connection.close();

            }

        });

    }



    //ctrl+o调出timeout，增加超时方法
    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) {
        System.out.println("超时的数据为：" + input);
    }
}
